package function;

import bean.createTableBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

import java.util.*;


public class richTableMapFunction extends RichMapFunction<createTableBean,String> {
    private ValueState<String> tableNameState;
    private ValueState<String> colState;
    Map<String,List<String>> m = new HashMap<String,List<String>>();
    List<String> l = new ArrayList<String>();
    String sql = "create table if not exists ";
    @Override
    public void close() throws Exception {
        Iterator<String> iterator = m.keySet().iterator();
        while(iterator.hasNext()){
            String key = iterator.next();
            List<String> list = m.get(key);
            sql = sql+key+"(";
            List<String> arrayList = new ArrayList<String>();
            for(int i=0;i<list.size();i++){
                String[] split = list.get(i).split("#");
                if(key.equals(split[0])) {
                    arrayList.add(split[1]);
                }
            }
            for(int i=0;i<arrayList.size();i++){
                if (i == arrayList.size() - 1) {
                    sql = sql + arrayList.get(i)+"\r"+")";
                } else {
                    sql = sql + arrayList.get(i) + ",";
                }
            }
            sql = sql+"PARTITIONED by (dt string COMMENT 'ETL日期') \n" +
                    "STORED AS parquet;";
            System.out.println(sql);
            sql="create table if not exists ";
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        tableNameState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
        colState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state2", String.class));
    }

    @Override
    public String map(createTableBean value) throws Exception {
        String v = tableNameState.value();
        if(v==null){
            tableNameState.update(value.getTableName());
            colState.update(value.getColName()+" "+value.getType()+" "+"comment '"+value.getComment()+"'"+"\n");
        }else{
            colState.update(value.getColName()+" "+value.getType()+" "+"comment '"+value.getComment()+"'"+"\n");
        }
        l.add(tableNameState.value()+"#"+colState.value());
        m.put(tableNameState.value(),l);
        return colState.value();
    }
}
